package com.cnpc.bds.sdk.sink.kafka.handler;

import com.cnpc.bds.sdk.sink.kafka.support.Assert;
import lombok.extern.slf4j.Slf4j;

/**
 * 责任链节点handler.
 *
 * <li>消息传递逻辑封装.
 */
@Slf4j
@SuppressWarnings({"rawtypes", "unchecked"})
public abstract class AbstractHandler<I, O> implements Handler<I, O> {

    private Handler next;
    private Handler prev;

    public Handler next() {
        return next;
    }

    public Handler prev() {
        return prev;
    }

    public void next(Handler next) {
        Assert.notNull(next, "The arg 'next' must be not null!");
        this.next = next;
    }

    public void prev(Handler prev) {
        Assert.notNull(prev, "The arg 'prev' must be not null!");
        this.prev = prev;
    }

    public void doHandle(I msg) {
        final O out = handle(msg);
        if (out != null) {
            fireNextHandle(out);
        }
    }

    public void fireNextHandle(O msg) {
        AbstractHandler next = (AbstractHandler) next();
        if (next == null) {
            return;
        }
        next.doHandle(msg);
    }
}
